package com.zhou;

import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import com.zhou.po.*;
import com.zhou.util.ESInputFormat;
import com.zhou.util.JDBCScript;
import com.zhou.util.PropertiesKey;
import org.apache.flink.api.common.functions.GroupReduceFunction;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.operators.Order;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.io.jdbc.JDBCInputFormat;
import org.apache.flink.api.java.io.jdbc.JDBCOutputFormat;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple7;
import org.apache.flink.api.java.typeutils.GenericTypeInfo;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.net.URL;
import java.sql.Timestamp;
import java.text.NumberFormat;
import java.time.LocalDate;
import java.time.ZoneOffset;
import java.util.List;
import java.util.Set;

public class CommunityChatBatchJob {

    private static final String PROPERTIES_NAME="community_chat.properties";

    private static final String SYS_MSG_1 = "您好，管家.*为您服务";

    private static final String SYS_MSG_2 = "您好，当前为管家休息时间";

    private static final Long HALF_AN_HOUR = 30 * 60 * 1000L;




    public static void main(String[] args) throws Exception {

       /* String path = CommunityChatBatchJob.class.getResource("").toString();
        System.out.println(path);*/
        URL path = CommunityChatBatchJob.class.getClassLoader().getResource("");
        System.out.println(path);
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        ParameterTool parameterTool = ParameterTool.fromPropertiesFile(path.getPath()+PROPERTIES_NAME);
        /*ESSource.initClient(parameterTool);
        ESSource esSource=new ESSource();
        esSource.run(null);*/

        JDBCInputFormat.JDBCInputFormatBuilder inputBuilder = JDBCInputFormat.buildJDBCInputFormat()
                .setDrivername(parameterTool.get(PropertiesKey.DASHBOARD_JOB_DRIVER_CLASS_NAME))
                .setDBUrl(parameterTool.get(PropertiesKey.DASHBOARD_JOB_JDBC_URL))
                .setUsername(parameterTool.get(PropertiesKey.DASHBOARD_JOB_USERNAME))
                .setPassword(parameterTool.get(PropertiesKey.DASHBOARD_JOB_PASSWORD))
                .setQuery(JDBCScript.DASHBOARD_JOB_QUERY)
                .setRowTypeInfo(JDBCScript.ROW_TYPE_INFO);
        DataSet<HdscCboardQuery> projectSource = env.createInput(inputBuilder.finish()).map(x -> {
            HdscCboardQuery query = new HdscCboardQuery();
            query.setCourtUuid((String) x.getField(0));
            query.setCourtName((String) x.getField(3));
            query.setRegionCompany((String) x.getField(1));
            query.setTotalHouseNumber(x.getField(5)==null?0:(int) x.getField(5));
            return query;
        });


        ESInputFormat.initClient(parameterTool);
        DataSet<CommunityChat> inputTasks = new DataSource(env,
                new ESInputFormat(),
                new GenericTypeInfo<>(CommunityChat.class), "ES").filter(x-> {
            CommunityChat chat=(CommunityChat)x;
            String msg = chat.getPayload().getBodies().get(0).getMsg();
            if (msg==null||msg.matches(SYS_MSG_1)||msg.matches(SYS_MSG_2)){
                return false;
            }
            return true;
        });
        inputTasks.print();

        List<String> userType= Lists.newArrayList("1","2","3","31","33");

        DataSet<ChatInProcess>
                chats =inputTasks.filter(x->{
            CommunityChat.Ext ext = x.getPayload().getExt();
            return userType.contains(ext.getFromUserType())&&userType.contains(ext.getToUserType());
        }).joinWithTiny(projectSource).where(new KeySelector<CommunityChat, String>() {
            @Override
            public String getKey(CommunityChat communityChat) throws Exception {
                String courtUuid = communityChat.getPayload().getExt().getCourtUuid();
                return courtUuid==null?"-1":courtUuid;
            }
        }).equalTo("courtUuid").with(new JoinFunction<CommunityChat, HdscCboardQuery, ChatInProcess>() {
            @Override
            public ChatInProcess join(CommunityChat chat, HdscCboardQuery row) throws Exception {
                String fromUserType = chat.getPayload().getExt().getFromUserType();
                CommunityChat.Ext ext = chat.getPayload().getExt();
                String sessionId = chat.getFrom() + "_" + chat.getTo();
                // 0表示业主发送给管家，1表示管家发送给业主
                Integer type = 0;
                if ("33".equals(fromUserType)) {
                    sessionId = chat.getTo() + "_" + chat.getFrom();
                    type = 1;
                }
                ChatInProcess fromChat = new ChatInProcess();
                fromChat.setRegion(row.getRegionCompany());
                fromChat.setCourtUuid(row.getCourtUuid());
                fromChat.setName(row.getCourtName());
                fromChat.setTotalHouseNumber(row.getTotalHouseNumber());
                String msg = chat.getPayload().getBodies().get(0).getMsg();
                fromChat.setMsg(msg.length() <= 1000 ? msg : msg.substring(0, 1000));
                fromChat.setSendTime(chat.getTimestamp());
                fromChat.setFromId(chat.getFrom());
                fromChat.setFromName(ext.getFromName());
                fromChat.setToId(chat.getTo());
                fromChat.setToName(ext.getToName());
                fromChat.setType(type);
                fromChat.setSessionId(sessionId);
                return fromChat;
            }
        });
        //全部聊天记录
            chats.map(x->{
                Row row=new Row(10);
                row.setField(0,x.getRegion());
                row.setField(1,x.getCourtUuid());
                row.setField(2,x.getName());
                row.setField(3,"2020-05-15");
                Timestamp timestamp=new Timestamp(x.getSendTime());
                row.setField(4,timestamp);
                row.setField(5,x.getMsg());
                row.setField(6,x.getFromId());
                row.setField(7,x.getFromName());
                row.setField(8,x.getType());
                row.setField(9,x.getSessionId());
                return row;
            }).output(JDBCOutputFormat.buildJDBCOutputFormat()
                    .setDrivername(parameterTool.get(PropertiesKey.BURIED_DRIVER_CLASS_NAME))
                    .setDBUrl(parameterTool.get(PropertiesKey.BURIED_JDBC_URL))
                    .setUsername(parameterTool.get(PropertiesKey.BURIED_USERNAME))
                    .setPassword(parameterTool.get(PropertiesKey.BURIED_PASSWORD))
                    .setQuery(JDBCScript.RESULT_CHART_INSERT)
                    .setSqlTypes(JDBCScript.SQL_TYPES)
                    .finish());
        //计算详情
        LocalDate date = LocalDate.now().minusDays(1);
        Long normalBegin = date.atTime(8, 0).toInstant(ZoneOffset.of("+8")).toEpochMilli();
        Long normalEnd = date.atTime(23, 0).toInstant(ZoneOffset.of("+8")).toEpochMilli();
        chats.print();
        DataSet<ChatComputeDetail>
                details =chats.groupBy("sessionId").sortGroup("sendTime", Order.ASCENDING).reduceGroup(new GroupReduceFunction<ChatInProcess, ChatComputeDetail>(){
            @Override
            public void reduce(Iterable<ChatInProcess> iterable, Collector<ChatComputeDetail> collector) throws Exception {
                List<Long> sessionReplayTimes = Lists.newArrayList();
                Long chatBeginTime = 0L;
                Long chatEndTime = 0L;
                int order=0;
                ChatInProcess first=null;
                for (ChatInProcess chat:iterable) {
                    if (0 == chat.getType() && chatBeginTime == 0) {
                        first=chat;
                        chatBeginTime = chat.getSendTime();
                    }
                    if (1 == chat.getType() && chatBeginTime > 0) {
                        chatEndTime = chat.getSendTime();
                    }
                    if (chatBeginTime > 0 && chatEndTime > 0) {
                        if (chatBeginTime < normalBegin) {
                            chatBeginTime = normalBegin;
                        } else if (chatBeginTime > normalEnd) {
                            chatBeginTime = 0l;
                            chatEndTime = 0l;
                            break;
                        }
                        if (chatEndTime < normalBegin) {
                            chatEndTime = normalBegin;
                        } else if (chatEndTime > normalEnd + HALF_AN_HOUR) {
                            chatEndTime = normalEnd + HALF_AN_HOUR;
                        }
                        long replayTime = chatEndTime - chatBeginTime;
                        sessionReplayTimes.add(replayTime);
                        ChatComputeDetail detail = new ChatComputeDetail();
                        detail.setRegion(chat.getRegion());
                        detail.setCommunityId(chat.getCourtUuid());
                        detail.setCommunityName(chat.getName());
                        detail.setStatTime("2020-05-15");
                        detail.setSessionId(chat.getSessionId());
                        detail.setStartTime(chatBeginTime);
                        detail.setEndTime(chatEndTime);
                        detail.setReplayTime(replayTime);
                        detail.setReplayOrder(order++);
                        detail.setTotalHouseNumber(chat.getTotalHouseNumber());
                        collector.collect(detail);
                        chatBeginTime = 0l;
                        chatEndTime = 0l;
                    }
                }
                if (chatBeginTime>=normalBegin&&chatBeginTime<=normalEnd&&chatEndTime==0){
                    ChatComputeDetail detail = new ChatComputeDetail();
                    detail.setRegion(first.getRegion());
                    detail.setCommunityId(first.getCourtUuid());
                    detail.setCommunityName(first.getName());
                    detail.setStatTime("2020-05-15");
                    detail.setSessionId(first.getSessionId());
                    detail.setStartTime(chatBeginTime);
                    detail.setEndTime(chatEndTime);
                    detail.setReplayTime(0l);
                    detail.setReplayOrder(order++);
                    detail.setTotalHouseNumber(first.getTotalHouseNumber());
                    collector.collect(detail);
                }
            }
        });
        //通过计算详情得到结果
        //项目
        DataSet<OnlineHousekeeperStatProject>
                projects =details.groupBy("communityId").reduceGroup(new GroupReduceFunction<ChatComputeDetail, OnlineHousekeeperStatProject>(){
            @Override
            public void reduce(Iterable<ChatComputeDetail> iterable, Collector<OnlineHousekeeperStatProject> collector) throws Exception {
                //1用户的个数
                Set<String> userIds = Sets.newHashSet();
                //2首次会话全部数量
                int firstTotal=0;
                //2.首次会话有回复个数
                int firstCount=0;
                //2首次响应时长总和
                long firstSum=0l;
                //2首次响应时长最大值
                long firstMax=0l;
                //4会话全部数量
                int total=0;
                //4会话有回复的个数
                int count=0;
                //4响应时长总和
                long sum=0l;
                //4响应时长最大值
                long max=0l;
                ChatComputeDetail first=null;
                for (ChatComputeDetail detail:iterable) {
                    if (first==null){
                        first=detail;
                    }
                    if (detail.getReplayOrder()==0l){
                        firstTotal++;
                    }
                    total++;
                    userIds.add(detail.getSessionId().split("_")[0]);
                    if (detail.getReplayTime()==0l){
                        continue;
                    }
                    if (detail.getReplayOrder()==0l){
                        firstCount++;
                        firstSum+=detail.getReplayTime();
                        if (firstMax<detail.getReplayTime()){
                            firstMax=detail.getReplayTime();
                        }
                    }
                    count++;
                    sum+=detail.getReplayTime();
                    if (max<detail.getReplayTime()){
                        max=detail.getReplayTime();
                    }
                }
                OnlineHousekeeperStatProject project=new OnlineHousekeeperStatProject();
                project.setRegion(first.getRegion());
                project.setCommunityId(first.getCommunityId());
                project.setProject(first.getCommunityName());
                project.setOwnerSendNums(userIds.size());
                project.setFirstResponseSum(firstSum);
                project.setFirstResponseCount(firstCount);
                project.setFirstTotal(firstTotal);
                project.setFirstResponseTimeAvg(firstCount==0?0:firstSum/firstCount);
                project.setFirstResponseTimeMax(firstMax);
                project.setResponseSum(sum);
                project.setResponseCount(count);
                project.setTotal(total);
                project.setResponseTimeAvg(count==0?0:sum/count);
                project.setResponseTimeMax(max);
                project.setTotalHouseNumber(first.getTotalHouseNumber());
                //回复率
                NumberFormat percent = NumberFormat.getPercentInstance();
                percent.setMinimumFractionDigits(2);
                BigDecimal divide = new BigDecimal(0);
                if (firstTotal > 0) {
                    divide = BigDecimal.valueOf(firstCount).divide(BigDecimal.valueOf(firstTotal), 4, BigDecimal.ROUND_HALF_UP);
                }
                String recoverRate = percent.format(divide.doubleValue());
                project.setRecoveryRate(recoverRate);
                //使用率
                BigDecimal useRate = new BigDecimal(0);
                if (first.getTotalHouseNumber() > 0) {
                    useRate = BigDecimal.valueOf(userIds.size()).divide(BigDecimal.valueOf(first.getTotalHouseNumber()), 4, BigDecimal.ROUND_HALF_UP);
                }
                percent.setMinimumFractionDigits(2);
                String useRateString = percent.format(useRate.doubleValue());
                project.setUseRate(useRateString);
                collector.collect(project);
            }
        });
        //地区
        DataSet<OnlineHousekeeperStatProject>
                regions =projects.groupBy("region").reduceGroup(new GroupReduceFunction<OnlineHousekeeperStatProject, OnlineHousekeeperStatProject>(){
            @Override
            public void reduce(Iterable<OnlineHousekeeperStatProject> iterable, Collector<OnlineHousekeeperStatProject> collector) throws Exception {
                OnlineHousekeeperStatProject region=null;
                for (OnlineHousekeeperStatProject detail:iterable) {
                    if (region==null){
                        region=detail;
                        region.setCommunityId(null);
                        region.setProject(null);
                        region.setFirstResponseTimeAvg(0l);
                        region.setFirstResponseTimeAvg(0l);
                        region.setRecoveryRate(null);
                        region.setUseRate(null);
                        continue;
                    }
                    region.setOwnerSendNums(region.getOwnerSendNums()+detail.getOwnerSendNums());
                    region.setFirstResponseSum(region.getFirstResponseSum()+detail.getFirstResponseSum());
                    region.setFirstResponseCount(region.getFirstResponseCount()+detail.getFirstResponseCount());
                    region.setFirstTotal(region.getFirstTotal()+detail.getFirstTotal());
                    if (region.getFirstResponseTimeMax()<detail.getFirstResponseTimeMax()){
                        region.setFirstResponseTimeMax(detail.getFirstResponseTimeMax());
                    }
                    region.setResponseSum(region.getResponseSum()+detail.getResponseSum());
                    region.setResponseCount(region.getResponseCount()+detail.getResponseCount());
                    region.setTotal(region.getTotal()+detail.getTotal());
                    if (region.getResponseTimeMax()<detail.getResponseTimeMax()){
                        region.setResponseTimeMax(detail.getResponseTimeMax());
                    }
                    region.setTotalHouseNumber(region.getTotalHouseNumber()+detail.getTotalHouseNumber());
                }
                region.setFirstResponseTimeAvg(region.getFirstResponseCount()==0?0:region.getFirstResponseSum()/region.getFirstResponseCount());
                region.setResponseTimeAvg(region.getResponseCount()==0?0:region.getResponseSum()/region.getResponseCount());
                //回复率
                NumberFormat percent = NumberFormat.getPercentInstance();
                percent.setMinimumFractionDigits(2);
                BigDecimal divide = new BigDecimal(0);
                if (region.getFirstTotal() > 0) {
                    divide = BigDecimal.valueOf(region.getFirstResponseCount()).divide(BigDecimal.valueOf(region.getFirstTotal()), 4, BigDecimal.ROUND_HALF_UP);
                }
                String recoverRate = percent.format(divide.doubleValue());
                region.setRecoveryRate(recoverRate);
                //使用率
                BigDecimal useRate = new BigDecimal(0);
                if (region.getTotalHouseNumber() > 0) {
                    useRate = BigDecimal.valueOf(region.getOwnerSendNums()).divide(BigDecimal.valueOf(region.getTotalHouseNumber()), 4, BigDecimal.ROUND_HALF_UP);
                }
                percent.setMinimumFractionDigits(2);
                String useRateString = percent.format(useRate.doubleValue());
                region.setUseRate(useRateString);
                collector.collect(region);
            }
        });
        //全国
        DataSet<OnlineHousekeeperStatProject>
                nation =regions.reduceGroup(new GroupReduceFunction<OnlineHousekeeperStatProject, OnlineHousekeeperStatProject>(){
            @Override
            public void reduce(Iterable<OnlineHousekeeperStatProject> iterable, Collector<OnlineHousekeeperStatProject> collector) throws Exception {
                OnlineHousekeeperStatProject region=null;
                for (OnlineHousekeeperStatProject detail:iterable) {
                    if (region==null){
                        region=detail;
                        region.setRegion("全国");
                        region.setCommunityId(null);
                        region.setProject(null);
                        region.setFirstResponseTimeAvg(0l);
                        region.setFirstResponseTimeAvg(0l);
                        region.setRecoveryRate(null);
                        region.setUseRate(null);
                        continue;
                    }
                    region.setOwnerSendNums(region.getOwnerSendNums()+detail.getOwnerSendNums());
                    region.setFirstResponseSum(region.getFirstResponseSum()+detail.getFirstResponseSum());
                    region.setFirstResponseCount(region.getFirstResponseCount()+detail.getFirstResponseCount());
                    region.setFirstTotal(region.getFirstTotal()+detail.getFirstTotal());
                    if (region.getFirstResponseTimeMax()<detail.getFirstResponseTimeMax()){
                        region.setFirstResponseTimeMax(detail.getFirstResponseTimeMax());
                    }
                    region.setResponseSum(region.getResponseSum()+detail.getResponseSum());
                    region.setResponseCount(region.getResponseCount()+detail.getResponseCount());
                    region.setTotal(region.getTotal()+detail.getTotal());
                    if (region.getResponseTimeMax()<detail.getResponseTimeMax()){
                        region.setResponseTimeMax(detail.getResponseTimeMax());
                    }
                    region.setTotalHouseNumber(region.getTotalHouseNumber()+detail.getTotalHouseNumber());
                }
                region.setFirstResponseTimeAvg(region.getFirstResponseCount()==0?0:region.getFirstResponseSum()/region.getFirstResponseCount());
                region.setResponseTimeAvg(region.getResponseCount()==0?0:region.getResponseSum()/region.getResponseCount());
                //回复率
                NumberFormat percent = NumberFormat.getPercentInstance();
                percent.setMinimumFractionDigits(2);
                BigDecimal divide = new BigDecimal(0);
                if (region.getFirstTotal() > 0) {
                    divide = BigDecimal.valueOf(region.getFirstResponseCount()).divide(BigDecimal.valueOf(region.getFirstTotal()), 4, BigDecimal.ROUND_HALF_UP);
                }
                String recoverRate = percent.format(divide.doubleValue());
                region.setRecoveryRate(recoverRate);
                //使用率
                BigDecimal useRate = new BigDecimal(0);
                if (region.getTotalHouseNumber() > 0) {
                    useRate = BigDecimal.valueOf(region.getOwnerSendNums()).divide(BigDecimal.valueOf(region.getTotalHouseNumber()), 4, BigDecimal.ROUND_HALF_UP);
                }
                percent.setMinimumFractionDigits(2);
                String useRateString = percent.format(useRate.doubleValue());
                region.setUseRate(useRateString);
                collector.collect(region);
            }
        });
        projects.print();
         projects.map(new ThansformTuple()).writeAsCsv("D:\\Users\\112469152\\workspace\\test-flink\\data\\csv1.csv" ).setParallelism(1);
        regions.map(new ThansformTuple()).writeAsCsv("D:\\Users\\112469152\\workspace\\test-flink\\data\\csv2.csv").setParallelism(1);
        nation.map(new ThansformTuple()).writeAsCsv("D:\\Users\\112469152\\workspace\\test-flink\\data\\csv3.csv").setParallelism(1);




//        List<ChatInProcess> collect = chats.collect();
//        chats.mapPartition(new MapPartitionFunction<ChatInProcess, Object>() {
//            @Override
//            public void mapPartition(Iterable<ChatInProcess> iterable, Collector<Object> collector) throws Exception {
//
//            }
//        })

        //System.out.println(collect);
        //projectSource.print();
        env.execute("CommunityChatBatchJob");

    }


    public static class ThansformTuple implements MapFunction<OnlineHousekeeperStatProject, Tuple7<String, String, String, Long, Long, String, String>> {

        @Override
        public Tuple7<String, String, String, Long, Long, String, String> map(OnlineHousekeeperStatProject onlineHousekeeperStatProject) throws Exception {
            Tuple7<String, String, String, Long, Long, String, String> tupLe=new Tuple7<>(onlineHousekeeperStatProject.getRegion()==null?"":onlineHousekeeperStatProject.getRegion(),
                    onlineHousekeeperStatProject.getCommunityId()==null?"":onlineHousekeeperStatProject.getCommunityId(),
                    onlineHousekeeperStatProject.getProject()==null?"":onlineHousekeeperStatProject.getProject(),
                    onlineHousekeeperStatProject.getFirstResponseTimeAvg()==null?0l:onlineHousekeeperStatProject.getFirstResponseTimeAvg(),
                    onlineHousekeeperStatProject.getResponseTimeAvg()==null?0l:onlineHousekeeperStatProject.getResponseTimeAvg(),
                    onlineHousekeeperStatProject.getUseRate()==null?"":onlineHousekeeperStatProject.getUseRate(),
                    onlineHousekeeperStatProject.getRecoveryRate()==null?"":onlineHousekeeperStatProject.getRecoveryRate()
                    );
            return tupLe;
        }
    }
}
